package com.hulk.rpc.discover.zk;

import com.hulk.rpc.config.common.AddressWithWeight;
import com.hulk.rpc.config.common.HostPort;
import com.hulk.rpc.config.server.ProtocolType;
import com.hulk.rpc.discover.Discover;
import com.hulk.rpc.discover.DiscoverListener;
import com.hulk.rpc.common.ForeverRetryPolicy;
import com.hulk.rpc.utils.concurrent.ConcurrentArrayList;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;

import java.io.IOException;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;

@Slf4j
public class ZooKeeperDiscover implements Discover {

  private CuratorFramework client;
  private ConcurrentArrayList<PathChildrenCache> watchers;

  @Override
  public void init(List<HostPort> hostPorts) {
    watchers = new ConcurrentArrayList<>();
    String connectString = hostPorts.stream().map(HostPort::toString)
            .collect(Collectors.joining(","));
    RetryPolicy retryPolicy = new ForeverRetryPolicy(1000, 60 * 1000);
    client = CuratorFrameworkFactory.newClient(connectString, 1000 * 10, 1000 * 3, retryPolicy);
    client.start();
  }

  @Override
  public void startRegistryListener(String group, String app, ProtocolType protocolType, DiscoverListener listener) {
    Objects.requireNonNull(listener, "listener is null");
    Objects.requireNonNull(client, "call init first");

    final String path = "/hulk/" + group + "/" + app + "/" + protocolType.name();

    final PathChildrenCache watcher = new PathChildrenCache(client, path, true);

    PathChildrenCacheListener pathChildrenCacheListener = new PathChildrenCacheListener() {
      private final ConcurrentMap<HostPort, Integer> serverWithWeight = new ConcurrentHashMap<>();
      private volatile boolean waitForInitializedEvent = true;

      @Override
      public void childEvent(CuratorFramework client, PathChildrenCacheEvent event)
              throws Exception {
        if (log.isInfoEnabled()) {
          log.info("zk监控列表发生变化, " + path + ", " + event.getType());
        }

        boolean isChanged = true;

        switch (event.getType()) {
          case INITIALIZED:
            waitForInitializedEvent = false;

            if (log.isInfoEnabled()) {
              log.info("完成初始化: " + path);
            }

            break;

          case CHILD_ADDED: {
            AddressWithWeight kv = new AddressWithWeight(event.getData().getData());
            serverWithWeight.put(kv.address, kv.weight);

            if (log.isInfoEnabled()) {
              log.info("新增节点: " + kv);
            }

            break;
          }

          case CHILD_REMOVED: {
            AddressWithWeight kv = new AddressWithWeight(event.getData().getData());
            serverWithWeight.remove(kv.address);

            if (log.isInfoEnabled()) {
              log.info("删除节点: " + kv);
            }

            break;
          }

          case CHILD_UPDATED: {
            AddressWithWeight kv = new AddressWithWeight(event.getData().getData());
            serverWithWeight.put(kv.address, kv.weight);

            if (log.isInfoEnabled()) {
              log.info("更新节点: " + kv);
            }

            break;
          }

          default:
            isChanged = false;

            if (log.isInfoEnabled()) {
              log.info("忽略, " + path + ", " + event.getType());
            }
        }

        if (!waitForInitializedEvent && isChanged) {
          try {
            listener.onChange(serverWithWeight);
          } catch (Throwable t) {
            if (log.isWarnEnabled()) {
              log.warn("Discover监听处理失败", t);
            }
          }
        }
      }
    };

    watcher.getListenable().addListener(pathChildrenCacheListener);

    try {
      watcher.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
      watchers.add(watcher);
    } catch (Exception e) {
      if (log.isErrorEnabled()) {
        log.error("zk监听失败, " + path, e);
      }
    }
  }

  @Override
  public void close() throws IOException {
    for (int i = 0; i < watchers.size(); i++) {
      PathChildrenCache watcher = watchers.get(i);

      try {
        watcher.close();
      } catch (Exception e) {
        if (log.isErrorEnabled()) {
          log.error("watcher关闭失败 ", e);
        }
      }
    }

    watchers = null;

    client.close();
    client = null;
  }
}
